package com.sh.data.engine.domain.integration.offline.service.impl;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.convert.Convert;
import cn.hutool.core.date.DateUtil;
import cn.hutool.core.util.StrUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.sh.data.engine.common.constants.Constants.LogFileExp;
import com.sh.data.engine.common.enumDefinition.DSType;
import com.sh.data.engine.common.enumDefinition.OfflineSyncTaskStatus;
import com.sh.data.engine.common.exception.BusinessException;
import com.sh.data.engine.common.util.ServletUtils;
import com.sh.data.engine.domain.authority.service.DataAuthService;
import com.sh.data.engine.domain.base.model.PageResult;
import com.sh.data.engine.domain.base.model.UserContext;
import com.sh.data.engine.domain.base.model.enums.DirectoryMenuEnum;
import com.sh.data.engine.domain.base.model.enums.RightEnum;
import com.sh.data.engine.domain.base.model.enums.XxlGroupEnum;
import com.sh.data.engine.domain.base.service.LogService;
import com.sh.data.engine.domain.common.service.DirectoryService;
import com.sh.data.engine.domain.common.service.DistributedService;
import com.sh.data.engine.domain.dataasset.model.domain.TableInfoBizDomain;
import com.sh.data.engine.domain.dataasset.model.param.QueryTableParam;
import com.sh.data.engine.domain.dataasset.service.TableService;
import com.sh.data.engine.domain.datadev.offline.model.enums.OfflineDevTaskStopEnum;
import com.sh.data.engine.domain.integration.datasource.model.domain.*;
import com.sh.data.engine.domain.integration.datasource.service.DataSourceService;
import com.sh.data.engine.domain.integration.offline.model.domain.*;
import com.sh.data.engine.domain.integration.offline.model.domain.OfflineTaskExistTableMappingDomain.Field;
import com.sh.data.engine.domain.integration.offline.model.domain.OfflineTaskExistTableMappingDomain.Table;
import com.sh.data.engine.domain.integration.offline.model.enums.OfflineSyncExecuTypeEnum;
import com.sh.data.engine.domain.integration.offline.model.mapper.OfflineSyncDatabaseDomainMapper;
import com.sh.data.engine.domain.integration.offline.model.mapper.OfflineSyncDomainMapper;
import com.sh.data.engine.domain.integration.offline.model.mapper.OfflineSyncRunningRecordDomainMapper;
import com.sh.data.engine.domain.integration.offline.model.mapper.OfflineSyncSimpleDomainMapper;
import com.sh.data.engine.domain.integration.offline.service.DataXService;
import com.sh.data.engine.domain.integration.offline.service.OfflineSyncRunningRecordService;
import com.sh.data.engine.domain.integration.offline.service.OfflineSyncService;
import com.sh.data.engine.domain.shims.db.BaseDbManager;
import com.sh.data.engine.domain.shims.db.model.FieldInfoDomain;
import com.sh.data.engine.domain.shims.db.model.TableInfoDomain;
import com.sh.data.engine.domain.shims.hbase.manager.HBaseManager;
import com.sh.data.engine.domain.shims.mongo.manager.MongoManager;
import com.sh.data.engine.domain.util.ConvertUtil;
import com.sh.data.engine.domain.workspace.manager.model.domain.UserDomain;
import com.sh.data.engine.domain.workspace.manager.service.UserService;
import com.sh.data.engine.infrastructure.config.FileStorageConfiguration;
import com.sh.data.engine.job.admin.controller.JobInfoController;
import com.sh.data.engine.job.admin.core.model.XxlJobInfo;
import com.sh.data.engine.job.admin.core.route.ExecutorRouteStrategyEnum;
import com.sh.data.engine.job.admin.core.scheduler.MisfireStrategyEnum;
import com.sh.data.engine.job.admin.core.scheduler.ScheduleTypeEnum;
import com.sh.data.engine.job.admin.service.XxlJobService;
import com.sh.data.engine.job.core.biz.model.ReturnT;
import com.sh.data.engine.job.core.enums.ExecutorBlockStrategyEnum;
import com.sh.data.engine.job.core.glue.GlueTypeEnum;
import com.sh.data.engine.repository.dao.common.entity.DirectoryEntity;
import com.sh.data.engine.repository.dao.integration.datasource.entity.DataSourceEntity;
import com.sh.data.engine.repository.dao.integration.offline.OfflineSyncMapper;
import com.sh.data.engine.repository.dao.integration.offline.entity.OfflineSyncEntity;
import com.sh.data.engine.repository.dao.integration.offline.entity.OfflineSyncEntity.OfflineSyncSchedule;
import com.sh.data.engine.repository.dao.integration.offline.entity.OfflineSyncRunningRecordEntity;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.io.IOException;
import java.time.LocalDateTime;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;

/**
 * @author binghe.yxh
 */
@Slf4j
@Service
public class OfflineSyncServiceImpl extends ServiceImpl<OfflineSyncMapper, OfflineSyncEntity>
    implements OfflineSyncService {

    @Value("${data.engine.offlineSyncHover:0}")
    private Integer offlineSyncHover;

    @Lazy
    @Autowired
    private DataSourceService dataSourceService;

    @Autowired
    private XxlJobService xxlJobService;

    @Autowired
    private JobInfoController jobInfoController;

    @Autowired
    private DataXService dataXService;

    @Autowired
    private UserService userService;

    @Autowired
    private DirectoryService directoryService;

    @Autowired
    private LogService logService;

    @Autowired
    private OfflineSyncRunningRecordService offlineSyncRunningRecordService;

    @Autowired
    private FileStorageConfiguration fileStorageConfiguration;

    @Autowired
    private DistributedService distributedService;

    @Autowired
    private DataAuthService dataAuthService;

    @Autowired
    private TableService tableService;

//    @Autowired
//    private TblPropService tblPropService;

//    @Autowired
//    private UserBehaviorLogService userBehaviorLogService;

    @Autowired
    private OfflineSyncDomainMapper offlineSyncDomainMapper;

    @Autowired
    private OfflineSyncRunningRecordDomainMapper offlineSyncRunningRecordDomainMapper;

    @Autowired
    private OfflineSyncDatabaseDomainMapper offlineSyncDatabaseDomainMapper;

    @Autowired
    private OfflineSyncSimpleDomainMapper offlineSyncSimpleDomainMapper;

    private static final int GP_TEXT = 1;

    private static final List<String> DS_TYPES =
        Arrays.asList(
            DSType.MySQL.name(),
            DSType.SQLServer.name(),
            DSType.PostgreSQL.name(),
            DSType.MatrixDB.name(),
            DSType.Oracle.name(),
            DSType.DB2.name(),
            DSType.Sap.name(),
            DSType.DM.name(),
            DSType.KINGBASE8.name(),
            DSType.Mongodb.name(),
            DSType.Hive.name(),
            DSType.HBase.name(),
            DSType.TiDB.name());

    /**
     * @param direction  0-来源 1-目标 允许写入当前数据库
     * @param sourceDsId 数据源id
     * @return List<OfflineSyncDatabaseDomain>
     */
    @Override
    public List<OfflineSyncDatabaseDomain> getOfflineSyncDatabases(
        Integer direction, String sourceDsId) {
        Long projectId = ServletUtils.getProjectId();
        List<String> dsTypes = CollUtil.newArrayList(DS_TYPES);
        if (direction.equals(1)) {
            if (StringUtils.isBlank(sourceDsId)) {
                throw new BusinessException("来源数据库id不能为空");
            }
            Long sourceId = Long.valueOf(sourceDsId);
            // TODO 目标数据库是否需要删除来源数据库
            DataSourceDetailDomain sourceDB = dataSourceService.getDataSourceDetailById(sourceId, false);
            if (sourceDB.isHive()) { // hive来源 目标关系型数据库
                dsTypes.remove(DSType.Mongodb.name());
                dsTypes.remove(DSType.HBase.name());
            }
            if (sourceDB.isMongoHbase()) { // 来源mongo 目标hive
                // dsTypes.clear();
                // dsTypes.add(DSType.Hive.name());
            }
            if (sourceDB.isRelatedDs()) { // 来源是关系型时 只选择hive和关系型
                dsTypes.remove(DSType.Mongodb.name());
                dsTypes.remove(DSType.HBase.name());
            }
        }
        DataSourceQueryDomain queryDomain =
            DataSourceQueryDomain.builder().dsTypes(dsTypes).projectId(projectId).build();
        List<DataSourceMultiDomain> dataSourceDomainList = dataSourceService.getDSByParams(queryDomain);
        dataSourceDomainList.sort(Comparator.comparing(DataSourceDomain::getDsType));
        return offlineSyncDatabaseDomainMapper.map(dataSourceDomainList);
    }

    @Override
    public Map<Long, String> getOfflineSncMapByIds(List<Long> ids) {
        List<OfflineSyncEntity> offlineSyncEntities = this.listByIds(ids);
        Map<Long, String> syncIdMap =
            offlineSyncEntities.stream()
                .collect(Collectors.toMap(OfflineSyncEntity::getId, OfflineSyncEntity::getTaskName));
        return syncIdMap;
    }

    /**
     * 编辑+保存离线同步任务
     *
     * @param request OfflineSyncDomain
     */
    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
//    @BizDataIndexAnnotation(
//        type = BizDataIndexTypeEnum.OFFLINE_TASK,
//        operateType = BizDataIndexAnnotation.OperateType.SAVE)
    public void saveOrUpdateOfflineSyncTask(OfflineSyncDomain request) {

        // 如果数据源是系统存储，需要判断密级
        Long sourceDsId = request.getSourceDsId();
        Long targetDsId = request.getTargetDsId();
        DataSourceEntity sourceEntity = dataSourceService.getById(sourceDsId);
        DataSourceEntity targetEntity = dataSourceService.getById(targetDsId);
//        boolean sourceS = true;
//        boolean targetS = true;
//        if (Objects.nonNull(sourceEntity.getIsSystem()) && sourceEntity.getIsSystem().equals(1)) {
//            sourceS =
//                tblPropService.checkSecretLevel(
//                    request.getSourceTable(), ServletUtils.getProjectId(), UserContext.getUserId());
//        }
//
//        if (Objects.nonNull(targetEntity.getIsSystem()) && targetEntity.getIsSystem().equals(1)) {
//            targetS =
//                tblPropService.checkSecretLevel(
//                    request.getSourceTable(), ServletUtils.getProjectId(), UserContext.getUserId());
//        }
//        if (!sourceS || !targetS) {
//            userBehaviorLogService.addFailedLog(
//                Business.离线同步任务, Operation.ADD_UPDATE, request.getTaskName(), LogResult.FAILED);
//            throw new BusinessException("密级不符合");
//        }
        OfflineSyncEntity offlineSyncEntity = Convert.convert(OfflineSyncEntity.class, request);
        offlineSyncEntity.setFieldMapping();
        List<OfflineSyncEntity.FieldMapping> fieldMappings = offlineSyncEntity.getFieldMappings();
        if (CollectionUtils.isEmpty(fieldMappings) && request.getEdit() == null) {
            throw new BusinessException("字段映射不能为空，请配置字段映射后重试！");
        }
        // 是否存在同名任务
        Long projectId = ServletUtils.getProjectId();
        LambdaQueryWrapper<OfflineSyncEntity> wrapper =
            new LambdaQueryWrapper<OfflineSyncEntity>()
                .eq(OfflineSyncEntity::getProjectId, projectId)
                .eq(OfflineSyncEntity::getTaskName, offlineSyncEntity.getTaskName());
        List<OfflineSyncEntity> sames = this.list(wrapper);
        Integer pgText = request.getPgText();
        if (Objects.nonNull(pgText) && pgText.equals(GP_TEXT)) {

            int fileNum = 0;
            // 数量匹配
            OfflineTaskExistTableMappingQueryDomain r1 = new OfflineTaskExistTableMappingQueryDomain();
            r1.setDsId(request.getTargetDsId());
            r1.setTables(Lists.newArrayList(request.getTargetTable()));
            OfflineTaskExistTableMappingDomain existTablesMappings = this.getExistTablesMappings(r1);
            if (Objects.nonNull(existTablesMappings)) {
                List<Table> tables = existTablesMappings.getTables();
                if (CollectionUtils.isNotEmpty(tables)) {
                    Table table = tables.get(0);
                    List<Field> fields = table.getFields();
                    fileNum = fields.size();
                }
            }
            // 字段匹配
            fieldMappings.forEach(
                e -> {
                    if (!e.getSourceField()
                        .toLowerCase(Locale.ROOT)
                        .equals(e.getTargetField().toLowerCase(Locale.ROOT))) {
                        throw new BusinessException(
                            "来源和目标字段映射的名称不相同，请重新编辑！");
                    }
                });
            if (fieldMappings.size() != fileNum) {
                throw new BusinessException("目标数据源的所有字段均需要连线，请重新编辑");
            }
        }
        if (Objects.isNull(request.getId())) { // 新增
            if (CollectionUtils.isNotEmpty(sames)) {
                throw new BusinessException("任务名称已存在，请检查后重试！");
            }
            offlineSyncEntity.setProjectId(projectId);
            offlineSyncEntity.setTaskStatus(OfflineSyncTaskStatus.torelease.getTaskStatus());
            // 如果是 文件写入模式需要 check下字段名称、类型、顺序一一匹配
            this.save(offlineSyncEntity);
//            LogContext.put(LogKey.NAME.getKey(), offlineSyncEntity.getTaskName());
        } else { // 编辑
            if (CollectionUtils.isNotEmpty(sames)
                && sames.stream()
                .anyMatch(
                    s ->
                        s.getTaskName().equals(offlineSyncEntity.getTaskName())
                            && !s.getId().equals(offlineSyncEntity.getId()))) {
                throw new BusinessException("任务名称已存在，请检查后重试！");
            }
            OfflineSyncEntity previousSyncEntity = this.getById(request.getId());
            if (Objects.isNull(previousSyncEntity)) {
                throw new BusinessException("离线同步任务不存在，请检查后重试！");
            }
            Long xxlId = offlineSyncEntity.getXxlJobId();
            offlineSyncEntity.setUpdateTime(new Date());
            offlineSyncEntity.setCreatorId(previousSyncEntity.getCreatorId());
            offlineSyncEntity.setCreateTime(previousSyncEntity.getCreateTime());
            offlineSyncEntity.setProjectId(previousSyncEntity.getProjectId());
            if (Objects.nonNull(xxlId)) {
                xxlJobService.remove(Math.toIntExact(xxlId));
            }
            offlineSyncEntity.setTaskStatus(OfflineSyncTaskStatus.torelease.getTaskStatus());
            this.updateById(offlineSyncEntity);
//            LogContext.put(LogKey.NAME.getKey(), offlineSyncEntity.getTaskName());
//            LogContext.put(LogKey.ISUPDATE.getKey(), LogKey.ISUPDATE.getValue());
        }
        // 任务状态未待发布
        if (request.getOpenSchedule().equals(1)) {
            String xxlJobId = this.saveXxlJob(offlineSyncEntity.getId(), false);
            offlineSyncEntity.setXxlJobId(Long.valueOf(xxlJobId));
            this.updateById(offlineSyncEntity);
        }
        request.setId(offlineSyncEntity.getId());
    }

    @Override
    @Transactional(rollbackFor = Exception.class, transactionManager = "engineTransactionManager")
    public String saveXxlJob(Long id, boolean reset) {

        OfflineSyncEntity offlineSyncEntity = this.getById(id);

        if (offlineSyncEntity == null) {
            return null;
        }

        OfflineSyncEntity.OfflineSyncSchedule schedule = offlineSyncEntity.getSchedule();

        if (schedule == null) {
            return null;
        }
        final UserDomain userInfo = UserContext.getUserInfo();

        int triggerStatus = 0;
        // 工作流未锁定且配置了调度,未发布
        if (Objects.nonNull(schedule.getStatus()) && schedule.getStatus().equals(0)) {
            triggerStatus = 1;
        }
        if (reset) {
            offlineSyncEntity.setXxlJobId(null);
        }
        boolean isInsert = true;
        XxlJobInfo jobInfo = new XxlJobInfo();
        if (Objects.nonNull(offlineSyncEntity.getXxlJobId())) {
            jobInfo.setId(Math.toIntExact(offlineSyncEntity.getXxlJobId()));
            isInsert = false;
        }
        // groupID
        jobInfo.setJobGroup(XxlGroupEnum.DATA_ENGINE_SYSTEMTASK_GROUP.getId());
        // 任务描述
        jobInfo.setJobDesc(offlineSyncEntity.getTaskName());
        // 责任人
        jobInfo.setAuthor(userInfo.getUserName());
        // 邮箱
        jobInfo.setAlarmEmail(userInfo.getEmail());
        // 调度执行器类型
        jobInfo.setScheduleType(ScheduleTypeEnum.HUFU.name());
        // 调度周期
        jobInfo.setScheduleConf(JSONObject.toJSONString(schedule));

        // 过期策略： 忽略
        jobInfo.setMisfireStrategy(MisfireStrategyEnum.DO_NOTHING.name());
        // 路由策略：轮询
        jobInfo.setExecutorRouteStrategy(ExecutorRouteStrategyEnum.ROUND.name());
        // 执行器
        jobInfo.setExecutorHandler("OfflineSyncHandler");
        // 阻塞策略： 单机串行
        jobInfo.setExecutorBlockStrategy(ExecutorBlockStrategyEnum.DISCARD_LATER.name());
        // 运行模式：Bean
        jobInfo.setGlueType(GlueTypeEnum.BEAN.name());
        jobInfo.setTriggerStatus(triggerStatus);
        jobInfo.setEffectFrom(schedule.getEffectiveDateStart());
        jobInfo.setEffectTo(schedule.getEffectiveDateEnd());

        if (isInsert) {
            final ReturnT<String> returnT = xxlJobService.add(jobInfo);

            final String content = returnT.getContent();

            if (StringUtils.isBlank(content)) {
                throw new BusinessException("xxlJob保存失败");
            }
            return content;
        }

        xxlJobService.update(jobInfo);
        return String.valueOf(offlineSyncEntity.getXxlJobId());
    }

    @Override
    public List<OfflineSyncDomain> getListByPage(Long projectId, Integer pageSize, Integer pageNum) {
        LambdaQueryWrapper<OfflineSyncEntity> wrapper =
            new LambdaQueryWrapper<OfflineSyncEntity>().eq(OfflineSyncEntity::getProjectId, projectId);
        Page<OfflineSyncEntity> page = this.page(Page.of(pageNum, pageSize), wrapper);
        return offlineSyncDomainMapper.map(page.getRecords());
    }

    @Override
    public List<UserDomain> getUserIds(Long projectId) {
        QueryWrapper<OfflineSyncEntity> queryChangeReasonWrapper =
            new QueryWrapper<OfflineSyncEntity>()
                .select("creator_id as creatorId")
                .eq("project_id", projectId)
                .isNotNull("creator_id")
                .groupBy("creator_id");
        List<Map<String, Object>> maps = listMaps(queryChangeReasonWrapper);

        log.info(maps.toString());
        List<String> userIds = Lists.newArrayList();
        maps.forEach(
            e -> {
                // pg
                Object creatorId = e.get("creatorid");
                if (creatorId == null) {
                    // dm
                    creatorId = e.get("CREATORID");
                    log.info("CREATORID:" + creatorId);
                }
                // mysql
                if (creatorId == null) {
                    creatorId = e.get("creatorId");
                }
                userIds.add(creatorId.toString());
            });

        List<UserDomain> convert = Lists.newArrayList();
        if (CollectionUtils.isNotEmpty(userIds)) {
            convert = Lists.newArrayList(userService.getMapUserDomainByUserIds(userIds).values());
        }
        return convert;
    }

    @Override
    public List<OfflineSyncEntity> getOfflineSyncByProjectId(Long projectId) {
        return this.list(
            new LambdaQueryWrapper<OfflineSyncEntity>().eq(OfflineSyncEntity::getProjectId, projectId));
    }

    @Override
    public List<String> getNodeTableById(Long jobId) {
        OfflineSyncEntity syncTaskByNodeId = this.getById(jobId);
        Long sourceDsId = syncTaskByNodeId.getSourceDsId();
        Long targetDsId = syncTaskByNodeId.getTargetDsId();
        List<String> strings = Lists.newArrayList();
        DataSourceDetailDomain dataSourceDetailById = dataSourceService.getDataSourceDetailById(
            sourceDsId, false);
        DataSourceDetailDomain targetSourceId = dataSourceService.getDataSourceDetailById(
            targetDsId, false);
        if (dataSourceDetailById.getIsSystem().equals(1)) {
            strings.add(syncTaskByNodeId.getSourceTable());
        } else if (targetSourceId.getIsSystem().equals(1)) {
            strings.add(syncTaskByNodeId.getTargetTable());
        }
        return strings;
    }

    private String saveOrUpdateXxlJob(Integer xxlJobId, OfflineSyncEntity offlineSyncEntity) {
        String jobId = "";
        if (Objects.isNull(xxlJobId)) {
            String glueSource = this.generateOfflineSyncGlueSource(offlineSyncEntity.getId());
            // 保存xxl-job
            XxlJobInfo xxlJobInfo = new XxlJobInfo();
            xxlJobInfo.setJobGroup(XxlGroupEnum.DATA_ENGINE_SYSTEMTASK_GROUP.getId());
            // TODO 重命名 xxlJobInfo.setJobDesc("离线同步_" + tm.getSourceTable() + "_" + tm.getTargetTable());
            xxlJobInfo.setJobDesc(offlineSyncEntity.getTaskName());
            xxlJobInfo.setAddTime(new Date());
            xxlJobInfo.setAuthor("1");

            xxlJobInfo.setScheduleType(ScheduleTypeEnum.HUFU.name());
            xxlJobInfo.setScheduleConf(JSON.toJSONString(offlineSyncEntity.getSchedule()));

            // 冻结状态
            xxlJobInfo.setTriggerStatus(0); // 默认是停止状态
            xxlJobInfo.setExecutorRouteStrategy(ExecutorRouteStrategyEnum.ROUND.name()); // 路由策略
            xxlJobInfo.setMisfireStrategy(MisfireStrategyEnum.DO_NOTHING.name()); // 调度过期策略

            // 阻塞处理策略
            Integer cycleDependency =
                Optional.ofNullable(offlineSyncEntity.getSchedule())
                    .map(OfflineSyncEntity.OfflineSyncSchedule::getCycleDependence)
                    .orElse(0);
            xxlJobInfo.setExecutorBlockStrategy(
                cycleDependency == 0
                    ? ExecutorBlockStrategyEnum.SERIAL_EXECUTION.name()
                    : ExecutorBlockStrategyEnum.DISCARD_LATER.name()); // 阻塞处理策略
            xxlJobInfo.setExecutorFailRetryCount(0); // 失败重试次数
            xxlJobInfo.setExecutorTimeout(0); // 任务超时时间
            xxlJobInfo.setGlueType(GlueTypeEnum.GLUE_GROOVY.name());
            xxlJobInfo.setGlueSource(glueSource); // glue source
            xxlJobInfo.setGlueUpdatetime(new Date());
            xxlJobInfo.setGlueRemark(LocalDateTime.now().toString());
            ReturnT<String> add = xxlJobService.add(xxlJobInfo);
            jobId = add.getContent();
        } else {
            XxlJobInfo jobInfo = xxlJobService.getById(xxlJobId);
            jobInfo.setJobDesc(offlineSyncEntity.getTaskName());
            jobInfo.setScheduleConf(JSON.toJSONString(offlineSyncEntity.getSchedule()));
            xxlJobService.update(jobInfo);
            jobId = String.valueOf(xxlJobId);
        }
        return jobId;
    }

    /**
     * 更新离线同步任务
     *
     * @param domain OfflineSyncDomain
     */
    @Override
    public void updateOfflineSyncTask(OfflineSyncDomain domain) {

        OfflineSyncEntity offlineSyncEntity = Convert.convert(OfflineSyncEntity.class, domain);
        this.updateById(offlineSyncEntity);
    }

    @Override
    public void updateOfflineSyncEndStatus(Long taskId, Integer taskRunStatus) {

        OfflineSyncEntity byId = this.getById(taskId);
        byId.setEndTime(new Date());
        byId.setTaskRunStatus(taskRunStatus);
        this.updateById(byId);
    }

    @Override
    public void stop(String jobIds) {
        List<Long> res =
            Arrays.stream(jobIds.split(",")).map(e -> Long.valueOf(e)).collect(Collectors.toList());

        for (Long jobId : res) {
            OfflineSyncRunningRecordEntity offlineSyncRunningRecordEntity =
                offlineSyncRunningRecordService.getById(jobId);
            Integer taskStatus = offlineSyncRunningRecordEntity.getTaskStatus();
            if (taskStatus.equals(OfflineSyncTaskStatus.stopped.getTaskStatus())
                || taskStatus.equals(OfflineSyncTaskStatus.failure.getTaskStatus())
                || taskStatus.equals(OfflineSyncTaskStatus.success.getTaskStatus())) {
                continue;
            }
            String ipAddress = offlineSyncRunningRecordEntity.getIpAddress();
            HashMap<String, String> paramMap = Maps.newHashMap();
            paramMap.put("logId", String.valueOf(jobId));
            boolean canStop =
                distributedService.stopTaskPost(
                    ipAddress, paramMap, OfflineDevTaskStopEnum.OFFLINE_SYNC.getCode());
            if (!canStop) {
                dataXService.stopOfflineSync(jobId);
            }
        }
    }

    @Override
    public String run(Long taskId, Integer runType) {
        return dataXService.executeOfflineSync(taskId, OfflineSyncExecuTypeEnum.MANUAL.getStatus());
    }

    @Override
    public List<OfflineSyncPageDomain> getSyncList(Long projectId, Long dirId) {

        List<Long> ids = Lists.newArrayList();
        // dirId为0，查询全部任务，不要很根据dirId以及下面的子级目录id去查，会导致没有定义目录的任务无法展示
        if (!"0".equals(dirId)) {
            List<DirectoryEntity> childrenById =
                directoryService.getChildrenById(dirId, projectId, DirectoryMenuEnum.OFFLINE);
            ids = childrenById.stream().map(DirectoryEntity::getId).collect(Collectors.toList());
            ids.add(dirId);
        }
        LambdaQueryWrapper<OfflineSyncEntity> in =
            new LambdaQueryWrapper<>(OfflineSyncEntity.class)
                .eq(OfflineSyncEntity::getProjectId, projectId)
                .in(OfflineSyncEntity::getDirId, ids);
        List<OfflineSyncEntity> list = list(in);
        List<String> userIds = list.stream().map(e -> e.getCreatorId()).collect(Collectors.toList());
        Map<String, String> userMap = userService.getMapByUserIds(new ArrayList<>(userIds));
        List<OfflineSyncPageDomain> offlineSyncPageDomains =
            Convert.toList(OfflineSyncPageDomain.class, list);
        offlineSyncPageDomains.forEach(
            e -> {
                e.setCreator(userMap.get(e.getCreatorId()));
            });

        return offlineSyncPageDomains;
    }

    //    @BizDataIndexAnnotation(
//        type = BizDataIndexTypeEnum.OFFLINE_TASK,
//        dataIdFieldName = "taskId",
//        operateType = BizDataIndexAnnotation.OperateType.SAVE)
    @Override
    public void online(Long taskId) {
        OfflineSyncEntity offlineSync = this.getById(taskId);
        offlineSync.setTaskStatus(OfflineSyncTaskStatus.released.getTaskStatus());
        offlineSync.setUpdateTime(new Date());
        offlineSync.setUpdaterId(Objects.requireNonNull(UserContext.getUserId()));
        offlineSync.setPublishTime(new Date());
        if (Objects.isNull(offlineSync.getSchedule()) || Objects.isNull(
            offlineSync.getSchedule().getEffectiveDateStart())) {
            throw new BusinessException("发布失败，请编辑补充调度信息");
        }
        this.updateById(offlineSync);
        jobInfoController.start(Math.toIntExact(offlineSync.getXxlJobId()));
    }

    //    @BizDataIndexAnnotation(
//        type = BizDataIndexTypeEnum.OFFLINE_TASK,
//        dataIdFieldName = "taskId",
//        operateType = BizDataIndexAnnotation.OperateType.SAVE)
    @Override
    public void offline(Long taskId) {
        OfflineSyncEntity offlineSync = this.getById(taskId);
        offlineSync.setTaskStatus(OfflineSyncTaskStatus.torelease.getTaskStatus());
        offlineSync.setUpdateTime(new Date());
        offlineSync.setUpdaterId(Objects.requireNonNull(UserContext.getUserId()));
        this.updateById(offlineSync);
        jobInfoController.stop(Math.toIntExact(offlineSync.getXxlJobId()));
    }

    @Override
    public PageResult<OfflineSyncRunningRecordDomain> historyTask(OfflineSyncTaskQueryDomain domain) {
        Long projectId = ServletUtils.getProjectId();
        IPage<OfflineSyncRunningRecordEntity> pageEntity =
            this.getRunningRecordPageList(
                domain.getPageNum(),
                domain.getPageSize(),
                domain.getTaskId(),
                domain.getTaskStatusList(),
                domain.getExecuteTypeList(),
                domain.getUserId(),
                domain.getStartTime(),
                domain.getEndTime(),
                projectId,
                domain.getOrder(),
                domain.getOrderField(),
                domain.getTaskName(),
                domain.getIgnored());

        List<OfflineSyncRunningRecordEntity> entityList = pageEntity.getRecords();
        PageResult<OfflineSyncRunningRecordDomain> pageDomain =
            PageResult.convert(
                pageEntity,
                res -> offlineSyncRunningRecordDomainMapper.entityToDomain(res));
        List<OfflineSyncRunningRecordDomain> domainList =
            Convert.toList(OfflineSyncRunningRecordDomain.class, entityList);

        List<String> res = domainList.stream().map(e -> e.getCreatorId()).collect(Collectors.toList());
        List<Long> syncIds =
            domainList.stream().map(e -> e.getOfflineSyncId()).collect(Collectors.toList());

        Map<String, String> mapByUserIds = Maps.newHashMap();
        if (CollectionUtils.isNotEmpty(res)) {
            mapByUserIds = userService.getMapByUserIds(res);
        }

        Map<String, String> finalMapByUserIds = mapByUserIds;

        domainList.forEach(
            d -> {
                String name = finalMapByUserIds.get(d.getCreatorId());
                d.setCreator(name);
                if (!Objects.isNull(d.getEndTime()) && !Objects.isNull(d.getStartTime())) {
                    long l = (d.getEndTime().getTime() - d.getStartTime().getTime()) / 1000;
                    d.setDuration(l);
                }
            });
        pageDomain.setResult(domainList);
        return pageDomain;
    }

    private Map<Long, OfflineSyncEntity> getMapIdName(List<Long> syncIds) {
        List<OfflineSyncEntity> offlineSyncEntities = this.listByIds(syncIds);
        Map<Long, OfflineSyncEntity> res =
            offlineSyncEntities.stream()
                .collect(Collectors.toMap(OfflineSyncEntity::getId, Function.identity()));
        return res;
    }

    private IPage<OfflineSyncRunningRecordEntity> getRunningRecordPageList(
        Integer pageNum,
        Integer pageSize,
        Long offlineSyncTaskId,
        String taskStatus,
        String executeType,
        String creatorId,
        Date startTime,
        Date endTime,
        Long projectId,
        String order,
        String orderField,
        String taskName,
        Integer ignored) {
        LambdaQueryWrapper<OfflineSyncRunningRecordEntity> wrapper =
            new LambdaQueryWrapper<OfflineSyncRunningRecordEntity>()
                .eq(
                    !Objects.isNull(creatorId),
                    OfflineSyncRunningRecordEntity::getCreatorId,
                    creatorId);
        if (Objects.nonNull(ignored)) {
            if (Objects.equals(ignored, 1)) {
                wrapper.eq(OfflineSyncRunningRecordEntity::getIgnored, 1);
            } else {
                wrapper.and(e -> e.eq(OfflineSyncRunningRecordEntity::getIgnored, 0).or()
                    .isNull(OfflineSyncRunningRecordEntity::getIgnored));
            }
        }

        if (StringUtils.isNotBlank(orderField) && StringUtils.isNotBlank(order)) {
            if (orderField.equalsIgnoreCase("plantime")) {
                wrapper.last("order by " + " plan_time " + order);
            } else if (orderField.equalsIgnoreCase("starttime")) {
                wrapper.last("order by " + " start_time " + order);
            } else if (orderField.equalsIgnoreCase("endtime")) {
                wrapper.last("order by " + " end_time " + order);
            } else if (orderField.equalsIgnoreCase("duration")) {
                wrapper.last("order by " + " (COALESCE(end_time,start_time) - start_time) " + order);
            }
        } else {
            wrapper.orderByDesc(OfflineSyncRunningRecordEntity::getStartTime);
        }
        wrapper.like(
            StringUtils.isNotBlank(taskName), OfflineSyncRunningRecordEntity::getTaskName, taskName);
        if (StringUtils.isNotBlank(taskStatus)) {
            String[] split = taskStatus.split(",");
            List<Integer> collect =
                Arrays.stream(split).map(e -> Integer.valueOf(e)).collect(Collectors.toList());
            wrapper.in(OfflineSyncRunningRecordEntity::getTaskStatus, collect);
        }

        if (StringUtils.isNotBlank(executeType)) {
            String[] split = executeType.split(",");
            List<Integer> collect =
                Arrays.stream(split).map(e -> Integer.valueOf(e)).collect(Collectors.toList());
            wrapper.in(OfflineSyncRunningRecordEntity::getExecuteType, collect);
        }

        wrapper.eq(
            Objects.nonNull(offlineSyncTaskId),
            OfflineSyncRunningRecordEntity::getOfflineSyncId,
            offlineSyncTaskId);
        if (Objects.nonNull(startTime)) {
            wrapper.ge(OfflineSyncRunningRecordEntity::getPlanTime, startTime);
        }
        if (Objects.nonNull(endTime)) {
            wrapper.le(OfflineSyncRunningRecordEntity::getPlanTime, endTime);
        }
        wrapper.eq(OfflineSyncRunningRecordEntity::getProjectId, projectId);
        return offlineSyncRunningRecordService.page(Page.of(pageNum, pageSize), wrapper);
    }

    //    @BizDataIndexAnnotation(
//        type = BizDataIndexTypeEnum.OFFLINE_TASK,
//        operateType = BizDataIndexAnnotation.OperateType.DELETE)
    @Override
    public Boolean deleteDir(Long id) {
        Long projectId = ServletUtils.getProjectId();
        List<DirectoryEntity> childrenById =
            directoryService.getChildrenById(
                id, projectId, DirectoryMenuEnum.OFFLINE);
        List<Long> ids = childrenById.stream().map(DirectoryEntity::getId).collect(Collectors.toList());
        ids.add(id);
        LambdaQueryWrapper<OfflineSyncEntity> wrapper =
            new LambdaQueryWrapper<OfflineSyncEntity>().in(OfflineSyncEntity::getDirId, ids);
        boolean flag = this.count(wrapper) == 0L;
        if (!flag) {
            throw new BusinessException("该目录下存在任务，无法删除");
        }
        directoryService.deleteDir(id, DirectoryMenuEnum.OFFLINE);
        return Boolean.TRUE;
    }

    @Override
    public List<OfflineSyncDomain> getOfflineTask(Long projectId, String tableName) {
        LambdaQueryWrapper<OfflineSyncEntity> wrapper =
            new LambdaQueryWrapper<OfflineSyncEntity>()
                .eq(OfflineSyncEntity::getProjectId, projectId)
                .eq(Objects.nonNull(tableName), OfflineSyncEntity::getTargetTable, tableName);
        List<OfflineSyncEntity> list = this.list(wrapper);
        return offlineSyncDomainMapper.map(list);
    }

    @Override
    public OfflineSyncLogDomain getOfflineSyncLog(Long logId, Integer offset) {
        int start = 1;
        if (offset != null && !offset.equals(0)) {
            start = offset;
        }
        OfflineSyncLogDomain response = new OfflineSyncLogDomain();
        response.setLogEnd(false);
        response.setOffset(offset);
        response.setLogContent(Lists.newArrayList());

        OfflineSyncRunningRecordEntity byId = offlineSyncRunningRecordService.getById(logId);
        if (byId == null) {
            ArrayList<String> log = Lists.newArrayList("该日志未找到");
            response.setLogContent(log);
            response.setLogEnd(true);
            return response;
        }
        if (!Objects.equals(OfflineSyncTaskStatus.running.getTaskStatus(), byId.getTaskStatus())) {
            response.setLogEnd(true);
        }
        // 仅仅只是读取单个子任务的日志
        List<String> logs;
        final String dir = fileStorageConfiguration.getDir();
        String subTaskJobLogPath =
            String.format("%s/log/%s/%s.log", dir, LogFileExp.OFFLINE_SYNC_LOG, byId.getId());

        try {
            logs = logService.read(subTaskJobLogPath, offset);
        } catch (IOException e) {
            log.error(e.getMessage());
            throw new BusinessException("读取日志异常");
        }
        if (CollectionUtils.isNotEmpty(logs)) {
            response.setOffset(logs.size() + start);
            response.setLogContent(logs);
        }
        return response;
    }

    //    @BizDataIndexAnnotation(
//        type = BizDataIndexTypeEnum.OFFLINE_TASK,
//        dataIdFieldName = "logId",
//        operateType = BizDataIndexAnnotation.OperateType.SAVE)
    @Override
    public void updateRunningRecord(Long logId) {
        OfflineSyncRunningRecordEntity runningRecordEntity =
            offlineSyncRunningRecordService.getById(logId);
        if (runningRecordEntity == null) {
            return;
        }
        Long offlineSyncId = runningRecordEntity.getOfflineSyncId();
        OfflineSyncEntity byId = this.getById(offlineSyncId);
        if (null == byId) {
            return;
        }
        byId.setTaskRunStatus(OfflineSyncTaskStatus.stopped.getTaskStatus());
        this.updateById(byId);
    }

    @Override
    public PageResult<OfflineSyncPageDomain> getPageList(OfflineSyncTaskQueryDomain domain) {
        Long projectId = domain.getProjectId();
        List<Long> ids = Lists.newArrayList();
        // dirId为0，查询全部任务，不要很根据dirId以及下面的子级目录id去查，会导致没有定义目录的任务无法展示
        if (Objects.nonNull(domain.getDirId()) && domain.getDirId() != 0) {
            List<DirectoryEntity> childrenById =
                directoryService.getChildrenById(
                    domain.getDirId(), domain.getProjectId(), DirectoryMenuEnum.OFFLINE);
            ids = childrenById.stream().map(DirectoryEntity::getId).collect(Collectors.toList());
            ids.add(domain.getDirId());
        }
        IPage<OfflineSyncEntity> pageList =
            this.getOfflineSyncPageList(
                domain.getPageNum(),
                domain.getPageSize(),
                domain.getTaskName(),
                domain.getTargetTableName(),
                domain.getTaskStatusList(),
                domain.getTaskRunStatus(),
                domain.getStartTime(),
                domain.getEndTime(),
                projectId,
                domain.getUserId(),
                domain.getSortTime(),
                ids,
                domain.getOrder(),
                domain.getOrderField(),
                domain.getCycle(),
                domain.getTargetTableNames());
        List<OfflineSyncEntity> records = pageList.getRecords();
        Map<Long, OfflineSyncEntity.OfflineSyncSchedule> map =
            records.stream()
                .collect(Collectors.toMap(OfflineSyncEntity::getId, OfflineSyncEntity::getSchedule));
        PageResult<OfflineSyncPageDomain> r1 =
            PageResult.copyPropertiesAndConvert(pageList, OfflineSyncPageDomain.class);
        List<OfflineSyncPageDomain> domainList = r1.getResult();
        if (CollectionUtils.isNotEmpty(domainList)) {
            // 数据源信息
            Set<Long> dsIds = new HashSet<>();
            Set<String> userIds = new HashSet<>();
            domainList.forEach(
                e -> {
                    dsIds.add(e.getSourceDsId());
                    dsIds.add(e.getTargetDsId());
                    userIds.add(e.getCreatorId());
                });
            List<DataSourceDomain> dataSourceDomains = dataSourceService.getDataSourceDomainByIds(dsIds);
            Map<Long, DataSourceDomain> dsDomainMap =
                dataSourceDomains.stream()
                    .collect(Collectors.toMap(DataSourceDomain::getId, d -> d, (d1, d2) -> d1));
            // 创建人信息
            Map<String, String> userMap = userService.getMapByUserIds(new ArrayList<>(userIds));
            domainList.forEach(
                d -> {
                    List<String> sources = Lists.newArrayList();
                    List<String> targets = Lists.newArrayList();
                    String sourceDsName =
                        Optional.ofNullable(dsDomainMap.get(d.getSourceDsId()))
                            .map(DataSourceDomain::getDsName)
                            .orElse("【数据源已删除】");
                    String targetDsName =
                        Optional.ofNullable(dsDomainMap.get(d.getTargetDsId()))
                            .map(DataSourceDomain::getDsName)
                            .orElse("【数据源已删除】");
                    sources.add(sourceDsName + "/" + d.getSourceTable());
                    targets.add(targetDsName + "/" + d.getTargetTable());
                    d.setSource(String.join(",", sources));
                    d.setTarget(String.join(",", targets));
                    if (d.getCreatorId() != null && userMap.containsKey(d.getCreatorId())) {
                        d.setCreator(userMap.get(d.getCreatorId()));
                    }
                    // 获取调度详情
                    OfflineSyncEntity.OfflineSyncSchedule offlineSyncSchedule = map.get(d.getId());
                    if (null != offlineSyncSchedule) {
                        String scheduleInfo = getScheduleInfo(offlineSyncSchedule);
                        d.setScheduleInfo(scheduleInfo);
                        d.setScheduleUnit(
                            Optional.ofNullable(offlineSyncSchedule)
                                .map(OfflineSyncEntity.OfflineSyncSchedule::getCycle)
                                .orElse(-1));
                    }
                });
        }
        return r1;
    }

    private IPage<OfflineSyncEntity> getOfflineSyncPageList(
        Integer pageNum,
        Integer pageSize,
        String taskName,
        String targetTableName,
        String taskStatus,
        Integer taskRunStatus,
        Date startTime,
        Date endTime,
        Long projectId,
        String creatorId,
        Integer sortTime,
        Collection<Long> dirIds,
        String order,
        String orderField,
        String cycle,
        Collection<String> targetTablesNames) {
        if (Objects.nonNull(endTime)) {
            endTime = DateUtil.endOfDay(endTime);
        }
        LambdaQueryWrapper<OfflineSyncEntity> wrapper =
            new LambdaQueryWrapper<OfflineSyncEntity>()
                .like(StrUtil.isNotBlank(taskName), OfflineSyncEntity::getTaskName, taskName)
                .like(
                    StrUtil.isNotBlank(targetTableName),
                    OfflineSyncEntity::getTargetTable,
                    targetTableName)
                //            .in(
                //                !Objects.isNull(taskRunStatus) && taskRunStatus.equals(3),
                //                OfflineSyncEntity::getTaskRunStatus,
                //                taskRunStatus,
                //                null)
                //            .eq(
                //                !Objects.isNull(taskRunStatus) && !taskRunStatus.equals(3),
                //                OfflineSyncEntity::getTaskRunStatus,
                //                taskRunStatus)
                .eq(!Objects.isNull(creatorId), OfflineSyncEntity::getCreatorId, creatorId)
                .ge(!Objects.isNull(startTime), OfflineSyncEntity::getEndTime, startTime)
                .le(!Objects.isNull(endTime), OfflineSyncEntity::getStartTime, endTime)
                .eq(Objects.nonNull(projectId), OfflineSyncEntity::getProjectId, projectId)
                .in(CollectionUtils.isNotEmpty(dirIds), OfflineSyncEntity::getDirId, dirIds)
                .in(
                    CollectionUtils.isNotEmpty(targetTablesNames),
                    OfflineSyncEntity::getTargetTable,
                    targetTablesNames);

        if (StringUtils.isNotBlank(cycle)) {
            String[] split = cycle.split(",");
            List<Integer> collect =
                Arrays.stream(split).map(e -> Integer.valueOf(e)).collect(Collectors.toList());
            //      wrapper.in(OfflineSyncEntity::)
        }

        if (StringUtils.isNotBlank(taskStatus)) {
            String[] split = taskStatus.split(",");
            List<Integer> collect =
                Arrays.stream(split).map(e -> Integer.valueOf(e)).collect(Collectors.toList());
            wrapper.in(OfflineSyncEntity::getTaskStatus, collect);
        }
        if (StringUtils.isNotBlank(order) && StringUtils.isNotBlank(orderField)) {
            if (orderField.equalsIgnoreCase("publishTime")) {
                wrapper.last(" order by publish_time " + order);
            }
        } else {
            wrapper.orderByDesc(OfflineSyncEntity::getUpdateTime);
        }

        return this.page(Page.of(pageNum, pageSize), wrapper);
    }

    private String getScheduleInfo(OfflineSyncSchedule schedule) {

        StringBuilder sb = new StringBuilder("执行时间: ");
        String startTimeHour = schedule.getStartTimeHour();
        String startTimeMin = schedule.getStartTimeMin();
        String endTimeHour = schedule.getEndTimeHour();
        String endTimeMin = schedule.getEndTimeMin();

        Integer offlineSyncHover = this.offlineSyncHover;
        if (offlineSyncHover.equals(1)) {
            if (StringUtils.isNotBlank(startTimeHour) && Integer.parseInt(startTimeHour) < 10) {
                startTimeHour = 0 + startTimeHour;
            }
            if (StringUtils.isNotBlank(startTimeMin) && Integer.parseInt(startTimeMin) < 10) {
                startTimeMin = 0 + startTimeMin;
            }
            if (StringUtils.isNotBlank(endTimeHour) && Integer.parseInt(endTimeHour) < 10) {
                endTimeHour = 0 + endTimeHour;
            }
            if (StringUtils.isNotBlank(endTimeMin) && Integer.parseInt(endTimeMin) < 10) {
                endTimeMin = 0 + endTimeMin;
            }
        }
        sb.append(startTimeHour);
        sb.append(":");
        sb.append(startTimeMin);
        Integer cycle = schedule.getCycle();
        if (cycle == null) {
            return null;
        }
        switch (cycle) {
            case 0:
                sb.append("\n执行间隔: ");
                sb.append("每月");
                sb.append(schedule.getExecPoint());
                sb.append("号");
                break;
            case 1:
                sb.append("\n执行间隔: ");
                sb.append("每周");
                sb.append(schedule.getExecPoint());
                break;
            case 2:
                break;
            case 3:
                sb.append("~");
                sb.append(endTimeHour);
                sb.append(":");
                sb.append(endTimeMin);
                sb.append("\n执行间隔: ");
                sb.append("每");
                sb.append(schedule.getExecPoint());
                sb.append("小时");
                break;
            case 4:
                sb.append("~");
                sb.append(endTimeHour);
                sb.append(":");
                sb.append(endTimeMin);
                sb.append("\n执行间隔: ");
                sb.append("每");
                sb.append(schedule.getExecPoint());
                sb.append("分钟");
                break;
        }
        return sb.toString();
    }

    @Override
    public List<OfflineSyncSimpleDomain> getOfflineSyncSimpleDomain(Long dsId, Long projectId) {
        LambdaQueryWrapper<OfflineSyncEntity> wrapper =
            new LambdaQueryWrapper<OfflineSyncEntity>()
                .eq(OfflineSyncEntity::getProjectId, projectId)
                .and(
                    w ->
                        w.eq(OfflineSyncEntity::getSourceDsId, dsId)
                            .or()
                            .eq(OfflineSyncEntity::getTargetDsId, dsId));
        List<OfflineSyncEntity> list = this.list(wrapper);
        return offlineSyncSimpleDomainMapper.map(list);
    }

    @Override
    public List<OfflineSyncSimpleDomain> getOfflineSyncListByProjectId(Long projectId) {
        LambdaQueryWrapper<OfflineSyncEntity> wrapper =
            new LambdaQueryWrapper<OfflineSyncEntity>()
                .eq(OfflineSyncEntity::getProjectId, projectId)
                .eq(OfflineSyncEntity::getTaskStatus, OfflineSyncTaskStatus.released.getTaskStatus())
                .orderByDesc(OfflineSyncEntity::getCreateTime);
        List<OfflineSyncEntity> list = this.list(wrapper);
        return offlineSyncSimpleDomainMapper.map(list);
    }

    @Override
    public OfflineSyncStatisticsDomain statistics() {
        Long projectId = ServletUtils.getProjectId();
        LambdaQueryWrapper<OfflineSyncEntity> wrapper =
            new LambdaQueryWrapper<OfflineSyncEntity>()
                .eq(Objects.nonNull(projectId), OfflineSyncEntity::getProjectId, projectId)
                .eq(OfflineSyncEntity::getTaskStatus, OfflineSyncTaskStatus.torelease.getTaskStatus());

        LambdaQueryWrapper<OfflineSyncRunningRecordEntity> eq =
            new LambdaQueryWrapper<OfflineSyncRunningRecordEntity>()
                .eq(OfflineSyncRunningRecordEntity::getProjectId, projectId)
//                .eq(OfflineSyncRunningRecordEntity::getExecuteType, 0)
                .eq(
                    OfflineSyncRunningRecordEntity::getTaskStatus,
                    OfflineSyncTaskStatus.success.getTaskStatus());
        long successCount = offlineSyncRunningRecordService.count(eq);

        LambdaQueryWrapper<OfflineSyncRunningRecordEntity> failue =
            new LambdaQueryWrapper<OfflineSyncRunningRecordEntity>()
                .eq(OfflineSyncRunningRecordEntity::getProjectId, projectId)
                .eq(
                    OfflineSyncRunningRecordEntity::getTaskStatus,
                    OfflineSyncTaskStatus.failure.getTaskStatus())
                .and(e -> e.eq(OfflineSyncRunningRecordEntity::getIgnored, 0).or()
                    .isNull(OfflineSyncRunningRecordEntity::getIgnored));
        long failueCount = offlineSyncRunningRecordService.count(failue);

        LambdaQueryWrapper<OfflineSyncRunningRecordEntity> runningWarpper =
            new LambdaQueryWrapper<OfflineSyncRunningRecordEntity>()
                .eq(OfflineSyncRunningRecordEntity::getProjectId, projectId)
                .eq(OfflineSyncRunningRecordEntity::getExecuteType, 1)
                .eq(
                    OfflineSyncRunningRecordEntity::getTaskStatus,
                    OfflineSyncTaskStatus.running.getTaskStatus());
        long runningCount = offlineSyncRunningRecordService.count(runningWarpper);
        Long release = this.count(wrapper);

        return OfflineSyncStatisticsDomain.builder()
            .toReleaseNum(release)
            .successNum(successCount)
            .failNum(failueCount)
            .runningNum(runningCount)
            .build();
    }

    @Override
    public OfflineSyncDomain getOfflineSyncById(Long id) {
        OfflineSyncEntity offlineSyncEntity = this.getById(id);
        if (Objects.isNull(offlineSyncEntity)) {
            throw new BusinessException("离线同步任务不存在，请检查后重试！");
        }
        OfflineSyncDomain offlineSyncDomain =
            Convert.convert(OfflineSyncDomain.class, offlineSyncEntity);
        DataSourceDetailDomain sourceDetail =
            dataSourceService.getDataSourceDetailById(offlineSyncDomain.getSourceDsId(), false);
        String sourceDSName =
            Optional.ofNullable(sourceDetail).map(DataSourceDomain::getDsName).orElse("数据源已删除");
        DataSourceDetailDomain targetDetail =
            dataSourceService.getDataSourceDetailById(offlineSyncDomain.getTargetDsId(), false);
        String targetDSName =
            Optional.ofNullable(targetDetail).map(DataSourceDomain::getDsName).orElse("数据源已删除");

        if (DSType.hasSchema(Objects.requireNonNull(sourceDetail).getDsType())) {
            String sourceTable = offlineSyncDomain.getSourceTable();
            if (sourceTable.contains(".")) {
                String[] split = sourceTable.split("\\.", 2);
                offlineSyncDomain.setSourceSchema(split[0]);
            }
        }

        if (DSType.hasSchema(Objects.requireNonNull(targetDetail).getDsType())) {
            String targetTable = offlineSyncDomain.getTargetTable();
            if (targetTable.contains(".")) {
                String[] split = targetTable.split("\\.", 2);
                offlineSyncDomain.setTargetSchema(split[0]);
            }
        }
        offlineSyncDomain.setSourceDsName(sourceDSName);
        offlineSyncDomain.setTargetDsName(targetDSName);
        return offlineSyncDomain;
    }

    @Override
    public OfflineSyncDomain getOfflineSyncByXxlId(Long xxlId) {
        LambdaQueryWrapper<OfflineSyncEntity> wrapper =
            new LambdaQueryWrapper<OfflineSyncEntity>().eq(OfflineSyncEntity::getXxlJobId, xxlId);
        OfflineSyncEntity one = this.getOne(wrapper);
        return Convert.convert(OfflineSyncDomain.class, one);
    }

    //    @BizDataIndexAnnotation(
//        type = BizDataIndexTypeEnum.OFFLINE_TASK,
//        operateType = BizDataIndexAnnotation.OperateType.DELETE)
    @Override
    public Boolean delOfflineSyncTask(Long id) {
        if (Objects.isNull(id)) {
            throw new BusinessException("离线任务id不能为空，请检查后重试！");
        }
        OfflineSyncEntity offlineSyncEntity = this.getById(id);
        if (Objects.isNull(offlineSyncEntity)) {
            throw new BusinessException("离线任务不存在，请检查后重试！");
        }

        Date before = new Date();

        if (Objects.nonNull(offlineSyncEntity.getXxlJobId())) {
            xxlJobService.remove(Math.toIntExact(offlineSyncEntity.getXxlJobId()));
        }
        Date after = new Date();
        long beforeTime = before.getTime();
        long afterTime = after.getTime();
        log.info("删除使用时间{}", (afterTime - beforeTime));
        this.removeById(id);

        Date after2 = new Date();
        long time = after2.getTime();
        log.info("删除任务使用时间{}", (time - afterTime));

        // 把运行记录也删除掉
        offlineSyncRunningRecordService.remove(
            new LambdaUpdateWrapper<OfflineSyncRunningRecordEntity>()
                .eq(OfflineSyncRunningRecordEntity::getOfflineSyncId, id));

        return Boolean.TRUE;
    }

    @Override
    public List<OfflineSyncTableDomain> getOfflineSyncTables(
        Integer direction, Long dsId, String schema) {
        DataSourceManagerDomain managerInfo = dataSourceService.getManagerInfo(dsId);
        if (Objects.isNull(managerInfo)) {
            throw new BusinessException("获取数据源信息失败，请检查后重试！");
        }
        List<TableInfoDomain> tableAndViewList = Lists.newArrayList();
        BaseDbManager baseDbManager = managerInfo.getDbManager();
        MongoManager mongoManager = managerInfo.getMongoManager();
        HBaseManager hBaseManager = managerInfo.getHBaseManager();
        String dbName = managerInfo.getDbName();
        try {
            if (!Objects.isNull(baseDbManager)) { // base
                // if (!DSType.Sap.name().equalsIgnoreCase(managerInfo.getDsType())) { // hana表数据量大，直接返回空
                tableAndViewList = baseDbManager.getTableAndViewList(dbName, schema);
                // }
            } else if (!Objects.isNull(mongoManager) || !Objects.isNull(hBaseManager)) { // mongo or hbase
                List<String> collectionNames = null;
                if (!Objects.isNull(mongoManager)) {
                    collectionNames = mongoManager.getCollectionList();
                } else {
                    collectionNames = hBaseManager.listCollectionNames();
                }
                tableAndViewList =
                    collectionNames.stream()
                        .map(
                            c -> {
                                TableInfoDomain tableInfo = new TableInfoDomain();
                                tableInfo.setTableName(c);
                                tableInfo.setTblType(TableInfoDomain.TblType.MANAGED_TABLE);
                                return tableInfo;
                            })
                        .collect(Collectors.toList());
            }
        } catch (Exception throwable) {
            log.error("离线同步，获取数据库表异常", throwable);
            throw new BusinessException(

                "离线同步，获取数据库[" + dbName + "]表异常，错误原因:" + throwable.getMessage()
                    + ",请检查数据库连接是否正常！");
        }

        // 对默认存储表权限处理
        String userId = UserContext.getUserId();
        Long projectId = ServletUtils.getProjectId();
        UserDomain userById = userService.getUserById(userId);
        if (Objects.nonNull(userById.getIsAuth()) && userById.getIsAuth().equals(1)) {
            return tableAndViewList.stream()
                .map(this::parseToOfflineSyncTableDomain)
                .sorted(Comparator.comparing(OfflineSyncTableDomain::getTableName))
                .collect(Collectors.toList());
        }
        Integer isSystem = managerInfo.getIsSystem();
        if (Objects.nonNull(isSystem) && isSystem.equals(1)) {
            Iterator<TableInfoDomain> iterator = tableAndViewList.iterator();
            List<TableInfoBizDomain> tableList =
                tableService.getTableList(projectId, new QueryTableParam());
            Map<String, Long> mapName =
                tableList.stream()
                    .collect(
                        Collectors.toMap(TableInfoBizDomain::getTableName, TableInfoBizDomain::getTblId));
            List<Long> tbInfoIds = tableList.stream().map(e -> e.getTblId()).collect(Collectors.toList());
            Map<Long, String> tblAuthMap = dataAuthService.getTblAuthMap(projectId, userId, tbInfoIds);
            while (iterator.hasNext()) {
                TableInfoDomain next = iterator.next();
                String tableName = next.getTableName();
                Long tblId = mapName.get(tableName);
                if (Objects.isNull(tblId)) {
                    continue;
                }
                String tblAuth = tblAuthMap.get(tblId);
                if (StringUtils.isBlank(tblAuth)) {
                    continue;
                }
                // 来源：无读写权限，过滤掉
                if (Objects.nonNull(direction)
                    && direction == 0
                    && RightEnum.NONE.getCode().equals(tblAuth)) {
                    iterator.remove();
                }
                // 目标：无写权限，过滤掉
                if (Objects.nonNull(direction)
                    && direction == 1
                    && !RightEnum.RW.getCode().equals(tblAuth)) {
                    iterator.remove();
                }
            }
        }
        return tableAndViewList.stream()
            .map(this::parseToOfflineSyncTableDomain)
            .sorted(Comparator.comparing(OfflineSyncTableDomain::getTableName))
            .collect(Collectors.toList());
    }

    /**
     * parse
     *
     * @param from TableInfoDomain
     * @return OfflineSyncTableDomain
     */
    private OfflineSyncTableDomain parseToOfflineSyncTableDomain(TableInfoDomain from) {
        return OfflineSyncTableDomain.builder()
            .tableName(from.getTableName())
            .isPartition(from.isPartitionTable())
            .schema(from.getSchema())
            .tblType(
                Optional.ofNullable(from.getTblType())
                    .map(TableInfoDomain.TblType::name)
                    .map(OfflineSyncTableDomain.TblType::valueOf)
                    .orElse(null))
            .detailedTblName(from.getTableComment())
            .build();
    }

    @Override
    public OfflineTaskExistTableMappingDomain getExistTablesMappings(
        OfflineTaskExistTableMappingQueryDomain request) {
        DataSourceManagerDomain managerInfo = dataSourceService.getManagerInfo(request.getDsId());
        if (Objects.isNull(managerInfo)) {
            throw new BusinessException("获取数据源信息失败，请检查后重试！");
        }
        BaseDbManager dbManager = managerInfo.getDbManager();
        MongoManager mongoManager = managerInfo.getMongoManager();
        List<String> tables = request.getTables();
        if (CollectionUtils.isEmpty(tables)) {
            return null;
        }
        if (Objects.isNull(dbManager)) {
            throw new BusinessException(
                "当前数据源类型不支持表字段查询，请手动输入表字段！");
        }
        List<OfflineTaskExistTableMappingDomain.Table> tableInfos = Lists.newArrayList();
        tables.forEach(
            e -> {
                try {
                    String dbName = managerInfo.getDbName();
                    TableInfoDomain tableInfo;
                    List<FieldInfoDomain> fieldList;
                    boolean flag = dbManager.hasSchema();
                    String tableName = e;
                    String schema = null;
                    if (flag && e.contains(".")) {
                        String[] split = e.split("\\.", 2);
                        tableName = split[1];
                        schema = split[0];
                    }
                    String dsType = managerInfo.getDsType();
                    if (dsType.equalsIgnoreCase(DSType.DM.name())) {
                        if (e.contains(".")) {
                            String[] split = e.split("\\.", 2);
                            tableName = split[1];
                            dbName = split[0];
                        }
                    }
                    if (StringUtils.isNotBlank(schema)) {
                        tableInfo = dbManager.getTableInfoByTableName(dbName, schema, tableName);
                        fieldList = dbManager.getFieldList(dbName, schema, tableName);
                    } else {
                        tableInfo = dbManager.getTableInfoByTableName(dbName, tableName);
                        fieldList = dbManager.getFieldList(dbName, tableName);
                    }

                    List<FieldInfoDomain> collect =
                        fieldList.stream()
                            .sorted(Comparator.comparing(FieldInfoDomain::getFieldName))
                            .collect(Collectors.toList());
                    List<FieldInfoDomain> partitionCollect =
                        fieldList.stream()
                            .filter(FieldInfoDomain::isPartition)
                            .sorted(Comparator.comparing(FieldInfoDomain::getFieldName))
                            .collect(Collectors.toList());

                    OfflineTaskExistTableMappingDomain.Table table =
                        ConvertUtil.copyProperties(
                            tableInfo, OfflineTaskExistTableMappingDomain.Table.class);
                    List<OfflineTaskExistTableMappingDomain.Field> fields =
                        ConvertUtil.copyProperties(
                            fieldList, OfflineTaskExistTableMappingDomain.Field.class);
                    List<OfflineTaskExistTableMappingDomain.Field> partitionFields =
                        ConvertUtil.copyProperties(
                            partitionCollect, OfflineTaskExistTableMappingDomain.Field.class);
                    table.setFields(fields);
                    table.setPartitionFields(partitionFields);
                    tableInfos.add(table);
                } catch (Exception throwable) {
                    log.error("离线同步，获取数据表字段异常：", throwable);
                    throw new BusinessException(

                        "获取表[" + e + "]字段信息异常，错误原因:" + throwable.getMessage());
                } finally {
                    dbManager.close();
                }
            });
        return OfflineTaskExistTableMappingDomain.builder().tables(tableInfos).build();
    }

    private String generateOfflineSyncGlueSource(Long taskId) {
        String glueJavaFormat =
            "package com.xuelangyun.hufu.job.jobhandler.offline;\n"
                + "\n"
                + "import com.xuelangyun.hufu.job.core.handler.IJobHandler;\n"
                + "import com.xuelangyun.hufu.service.integration.offline.DataXService;\n"
                + "import org.springframework.beans.factory.annotation.Autowired;\n"
                + "\n"
                + "public class OfflineSyncDemoHandler extends IJobHandler {\n"
                + "\n"
                + "  // 离线同步id\n"
                + "  private String offlineSyncId = \"%s\";\n"
                + "\n"
                + "  @Autowired private DataXService dataXService;\n"
                + "\n"
                + "  @Override\n"
                + "  public void execute() throws Exception {\n"
                + "    dataXService.executeOfflineSync(offlineSyncId);\n"
                + "  }\n"
                + "\n"
                + "  @Override\n"
                + "  public void destroy(){\n"
                + "    dataXService.destroy(offlineSyncId);\n"
                + "  }\n"
                + "}\n";
        return String.format(glueJavaFormat, taskId);
    }
}
